use std::io::{self, Cursor};

use bytes::{Buf, BytesMut};
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufWriter};
use tokio::net::TcpStream;

use crate::frame::{self, Frame};

#[derive(Debug)]
pub struct Connection {
    // TcpStream：包装成 BufWriter，提供读取和写出数据
    stream: BufWriter<TcpStream>,
    // 缓冲区：用来读取Stream中数据
    buffer: BytesMut,
}


impl Connection {
    pub fn new(socket: TcpStream) -> Connection {
        Connection {
            stream: BufWriter::new(socket),
            // BytesMut可以自动扩容
            buffer: BytesMut::with_capacity(32),
        }
    }

    /// 读取数据帧
    pub async fn read_frame(&mut self) -> crate::Result<Option<Frame>> {
        loop {
            // 解析帧：尝试从缓冲数据中解析帧，如果缓冲了足够的数据，则返回帧
            // 没有缓存到足够的数据，parse_frame返回的是None,则会进入下面的流程继续读取
            if let Some(frame) = self.parse_frame()? {
                return Ok(Some(frame));
            }

            // stream.read_buf 从socket中读取数据，返回读取的bytes数量
            // 如果返回0，代表 stream 结束
            // self.buffer可以自动扩容：不用担心数据过多超出缓冲区容量
            if 0 == self.stream.read_buf(&mut self.buffer).await? {
                if self.buffer.is_empty() {
                    // 缓冲区没有数据，正常关闭
                    return Ok(None);
                } else {
                    // 缓冲区有数据，这意味着对等端在发送帧时关闭了socket
                    return Err("connection reset by peer".into());
                }
            }
            // 注意：如果流没有结束，进入下一次循环
        }
    }

    // 解析帧
    // 如果缓冲区包含足够的数据，则返回帧并从缓冲区中删除数据
    // 如果尚未缓冲足够的数据，则返回 Ok(None)
    // 如果缓冲数据不表示有效帧，则返回 Err
    fn parse_frame(&mut self) -> crate::Result<Option<Frame>> {
        use frame::Error::Incomplete;
        // Cursor 跟踪缓冲区中的“当前”位置
        let mut buf = Cursor::new(&self.buffer[..]);

        // 检查是否缓冲了足够的数据来解析单个帧
        match Frame::check(&mut buf) {
            // check 成功
            Ok(_) => {
                // check() 将光标前移到帧的末尾,通过光标位置来获得帧的长度
                let len = buf.position() as usize;
                // 重置为0
                buf.set_position(0);

                // 解析缓冲区中的帧，并返回帧值
                // 如果编码帧表示无效，则返回错误
                // 这将终止当前连接，但不会影响任何其他已连接的客户端
                let frame = Frame::parse(&mut buf)?;

                // 放弃读取缓冲区中的解析数据，丢弃 len 之前的所有数据
                self.buffer.advance(len);
                Ok(Some(frame))
            }
            // 读取缓冲区中的数据不足，无法解析单个帧，返回None
            Err(Incomplete) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// 依赖Frame协议
    pub async fn write_frame(&mut self, frame: &Frame) -> io::Result<()> {
        match frame {
            // array ： * ｜len | entry1 | entry2
            Frame::Array(val) => {
                self.stream.write_u8(b'*').await?;
                self.write_decimal(val.len() as u64).await?;
                for entry in &**val {
                    self.write_value(entry).await?;
                }
            }
            _ => self.write_value(frame).await?,
        }
        self.stream.flush().await
    }

    /// 依赖Frame 协议
    async fn write_value(&mut self, frame: &Frame) -> io::Result<()> {
        match frame {
            Frame::Simple(val) => {
                self.stream.write_u8(b'+').await?;
                self.stream.write_all(val.as_bytes()).await?;
                self.stream.write_all(b"\r\n").await?;
            }
            Frame::Error(val) => {
                self.stream.write_u8(b'-').await?;
                self.stream.write_all(val.as_bytes()).await?;
                self.stream.write_all(b"\r\n").await?;
            }
            Frame::Integer(val) => {
                self.stream.write_u8(b':').await?;
                self.write_decimal(*val).await?;
            }
            Frame::Null => {
                self.stream.write_all(b"$-1\r\n").await?;
            }
            Frame::Bulk(val) => {
                let len = val.len();

                self.stream.write_u8(b'$').await?;
                self.write_decimal(len as u64).await?;
                self.stream.write_all(val).await?;
                self.stream.write_all(b"\r\n").await?;
            }
            Frame::Array(_val) => unreachable!(),
        }

        Ok(())
    }

    async fn write_decimal(&mut self, val: u64) -> io::Result<()> {
        use std::io::Write;

        // 转换成string
        let mut buf = [0u8; 20];
        let mut buf = Cursor::new(&mut buf[..]);
        write!(&mut buf, "{}", val)?;

        let pos = buf.position() as usize;
        self.stream.write_all(&buf.get_ref()[..pos]).await?;
        self.stream.write_all(b"\r\n").await?;

        Ok(())
    }
}